package com.nl.orderpay;

import com.nl.bean.input.OrderEvent;
import com.nl.bean.input.ReceiptEvent;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author shihb
 * @date 2019/12/27 10:17
 * 对账类，联合多条流 一条订单支付，
 */
public class TrByJoin {


  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    //1.订单事件流
    DataStreamSource<String> paySource = env.readTextFile(
        TrByJoin.class.getClassLoader().getResource("OrderLog.csv").getPath());
    //2.交易回执流
    DataStreamSource<String> receiptSource = env.readTextFile(
        TrByJoin.class.getClassLoader().getResource("ReceiptLog.csv").getPath());

    KeyedStream<OrderEvent, String> payStream = paySource
        .map(s -> {
          String[] arr = s.split(",");
          return new OrderEvent(Long.parseLong(arr[0].trim()), arr[1].trim(), arr[2].trim(),
              Long.parseLong(arr[3].trim()));
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<OrderEvent>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(OrderEvent element) {
                return element.getEventTime() * 1000;
              }
            }
        )
        .filter(orderEvent -> !"".equals(orderEvent.getTransactionId()))
        .keyBy(orderEvent -> orderEvent.getTransactionId());

    KeyedStream<ReceiptEvent, String> receiptStream = receiptSource
        .map(s -> {
          String[] arr = s.split(",");
          return new ReceiptEvent(arr[0].trim(), arr[1].trim(), Long.parseLong(arr[2].trim()));
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<ReceiptEvent>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(ReceiptEvent element) {
                return element.getReceiptTime() * 1000;
              }
            }
        ).keyBy(ReceiptEvent -> ReceiptEvent.getTransactionId());

    // 将两条流连接起来interval用来处理两条流匹配上的情况,本案例并不是很适合
    SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> trStream = payStream
        .intervalJoin(receiptStream).between(Time.seconds(-5), Time.seconds(5))
        .process(new TrFunction());

    trStream.print("success");
    env.execute("tr job");

  }

  private static class TrFunction extends
      ProcessJoinFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {

    @Override
    public void processElement(OrderEvent left, ReceiptEvent right, Context ctx,
        Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
      out.collect(new Tuple2<>(left, right));
    }
  }
}


